package com.gph.saviorframework.mq.api.impl.rocketmq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.gph.saviorframework.mq.api.*;
import com.gph.saviorframework.mq.api.exception.MQClientException;
import com.gph.saviorframework.mq.api.impl.util.MQUtil;

import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by root on 16-7-10.
 */
public class ConsumerImpl extends MQClientAbstract implements Consumer {

    private final DefaultMQPushConsumer defaultMQPushConsumer;

    private final ConcurrentHashMap<String/* Topic */, MessageListener> subscribeTable = new ConcurrentHashMap<String, MessageListener>();

    private final AtomicBoolean started = new AtomicBoolean(false);

    public ConsumerImpl(final Properties properties){

        super(properties);

        this.defaultMQPushConsumer = new DefaultMQPushConsumer(new ClientRPCHook(sessionCredentials));

        String consumerGroup = properties.getProperty(PropertyKeyConst.ConsumerId);
        if (null == consumerGroup) {
            throw new MQClientException("\'ConsumerId\' property is null");
        }

        String maxReconsumeTimes = properties.getProperty(PropertyKeyConst.MaxReconsumeTimes);
        if (!UtilAll.isBlank(maxReconsumeTimes)) {
            try {
                this.defaultMQPushConsumer.setMaxReconsumeTimes(Integer.parseInt(maxReconsumeTimes));
            }
            catch (NumberFormatException e) {
                e.printStackTrace();
            }
        }

        String consumeTimeout = properties.getProperty(PropertyKeyConst.ConsumeTimeout);
        if (!UtilAll.isBlank(consumeTimeout)) {
            try {
                this.defaultMQPushConsumer.setConsumeTimeout(Integer.parseInt(consumeTimeout));
            }
            catch (NumberFormatException e) {
                e.printStackTrace();
            }
        }

        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        this.defaultMQPushConsumer.setVipChannelEnabled(isVipChannelEnabled);

        String messageModel = properties.getProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        this.defaultMQPushConsumer.setMessageModel(MessageModel.valueOf(messageModel));//集群消费
        this.defaultMQPushConsumer.setConsumerGroup(consumerGroup);
        this.defaultMQPushConsumer.setInstanceName(this.buildIntanceName());
        this.defaultMQPushConsumer.setNamesrvAddr(this.nameServerAddr);

        if (properties.containsKey(PropertyKeyConst.ConsumeThreadNums)) {
            this.defaultMQPushConsumer.setConsumeThreadMin(Integer.valueOf(properties.get(PropertyKeyConst.ConsumeThreadNums).toString()));
            this.defaultMQPushConsumer.setConsumeThreadMax(Integer.valueOf(properties.get(PropertyKeyConst.ConsumeThreadNums).toString()));
        }
    }

    /**
     * 启动服务
     */
    public void start() {
        this.defaultMQPushConsumer.registerMessageListener(new MessageListenerImpl());
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQPushConsumer.start();
            }
        } catch (Exception e) {
            throw new MQClientException(e.getMessage());
        }
    }

    /**
     * 关闭服务
     */
    public void shutdown() {
        if (this.started.compareAndSet(true, false)) {
            this.defaultMQPushConsumer.shutdown();
        }
    }

    /**
     * 订阅消息
     *
     * @param topic         消息主题
     * @param subExpression 订阅过滤表达式字符串，ONS服务器依据此表达式进行过滤。只支持或运算<br>
     *                      eg: "tag1 || tag2 || tag3"<br>
     *                      如果subExpression等于null或者*，则表示全部订阅
     * @param listener
     */
    public void subscribe(String topic, String subExpression, MessageListener listener) {
        if (null == topic) {
            throw new MQClientException("topic is null");
        }

        if (null == listener) {
            throw new MQClientException("listener is null");
        }

        try {
            this.subscribeTable.put(topic, listener);
            this.defaultMQPushConsumer.subscribe(topic, subExpression);
        }
        catch (com.alibaba.rocketmq.client.exception.MQClientException e) {
            throw new MQClientException("defaultMQPushConsumer subscribe exception", e);
        }
    }

    /**
     * 取消某个topic订阅
     *
     * @param topic
     */
    public void unsubscribe(String topic) {
        if (null != topic) {
            this.defaultMQPushConsumer.unsubscribe(topic);
        }
    }

    public boolean isStarted() {
        return started.get();
    }

    public boolean isClosed() {
        return !isStarted();
    }

    class MessageListenerImpl implements MessageListenerConcurrently {

        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgsRMQList, ConsumeConcurrentlyContext contextRMQ) {
            MessageExt msgRMQ = msgsRMQList.get(0);
            Message msg = MQUtil.msgConvert(msgRMQ);
            Map<String, String> stringStringMap = msgRMQ.getProperties();
            msg.setMsgID(msgRMQ.getMsgId());
            if (stringStringMap != null && stringStringMap.get(Constants.TRANSACTION_ID) != null) {
                msg.setMsgID(stringStringMap.get(Constants.TRANSACTION_ID));
            }
            MessageListener listener = ConsumerImpl.this.subscribeTable.get(msg.getTopic());
            if (null == listener) {
                throw new MQClientException("MessageListener is null");
            }

            final ConsumeContext context = new ConsumeContext();
            Action action = listener.consume(msg, context);
            if (action != null) {
                switch (action) {
                    case CommitMessage:
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    case ReconsumeLater:
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    default:
                        break;
                }
            }

            return null;
        }
    }

}
